package com.haiyou.data.common.accessor.redis.redisson.bucket;

import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.BiFunction;
import java.util.function.Function;
import java.util.stream.Collectors;

import org.redisson.api.BatchOptions;
import org.redisson.api.RBatch;

import com.google.common.collect.Iterables;
import com.google.common.collect.Sets;
import com.haiyou.common.entity.Entity;
import com.haiyou.data.common.accessor.redis.redisson.RedissonAccessor;
import com.haiyou.data.common.accessor.rsync.Synchronizer;
import com.haiyou.data.common.cache.redis.RKeys;
import com.haiyou.data.common.cache.redis.redisson.RedissonAtomicExecutor;
import com.haiyou.data.common.persister.mysql.persister.Persister;

import lombok.Getter;
import lombok.extern.slf4j.Slf4j;

/**
 * 使用redisson对象桶操作数据, 并同步到mysql
 * </p>
 * 
 * @author xingyuan
 *
 * @param <V>
 */
@Slf4j
public abstract class RedissonBucketAccessor<V extends Entity<String>> extends RedissonAccessor<String, V> {
	@Getter
	protected Persister<String, V> persister;
	@Getter
	protected Synchronizer<String, V> synchronizer;

	@Override
	public void init() {
		super.init();
		persister = Persister.getInstance(valueClass);
		synchronizer = new Synchronizer<>(this, persister);
		synchronizer.init();
		timeToLive = synchronizer.getTimeToLive();
		timeToLiveUnit = synchronizer.getTimeToLiveUnit();
	}

	@Override
	public void destory() {
		super.destory();
		synchronizer.destory();
	}

//	@Override
//	public Map<String, V> getAll(Iterable<? extends String> keys) {
//		Set<String> keySet = Sets.newHashSet(keys);
//		Map<String, V> result = redissonClient.getBuckets().get(Iterables.toArray(keys, keyClass));
//		keySet.removeAll(result.keySet());
//		if (!keySet.isEmpty()) {
//			RLock[] locks = new RLock[keySet.size()];
//			int i = 0;
//			for (String key : keySet) {
//				locks[i++] = RedissonAtomicExecutor.getLock(redissonClient, key);
//			}
//			RLock lock = RedissonAtomicExecutor.getMultiLock(redissonClient, locks);// TODO
//																					// 联锁
//			try {
//				if (lock.tryLock(lockWaitTime(), lockLeaseTime(), lockTimeUnit())) {
//					Set<String> queryKeys = Sets.newHashSet();// 需要从db获取
//					Set<String> reGetKeys = Sets.newHashSet();// 需要再次从redis获取
//					for (String key : keySet) {
//						if (!redissonClient.getBucket(key).isExists()) {
//							queryKeys.add(key);
//						} else {
//							reGetKeys.add(key);
//						}
//					}
//					if (!queryKeys.isEmpty()) {
//						List<V> queryList = persister.queryBatch(queryKeys);
//						if (!queryList.isEmpty()) {
//							RBatch batch = redissonClient.createBatch(BatchOptions.defaults());
//							result.putAll(queryList.stream().collect(Collectors.toMap(V::getId, n -> {
//								if (timeToLive > 0) {
//									batch.getBucket(n.getId()).trySetAsync(n, timeToLive, timeToLiveUnit);
//								} else {
//									batch.getBucket(n.getId()).trySetAsync(n);
//								}
//								return n;
//							}, (n, m) -> m)));
//							batch.execute();
//						}
//					}
//					if (!reGetKeys.isEmpty()) {
//						Map<String, V> reGetMap = redissonClient.getBuckets()
//								.get(Iterables.toArray(reGetKeys, getKeyClass()));
//						result.putAll(reGetMap);
//					}
//				} else {
//					log.error(
//							"没有获取到分布式联锁:" + valueClass + "," + Arrays.toString(Iterables.toArray(keySet, String.class)),
//							new Throwable());
//				}
//			} catch (Exception e) {
//				log.error(e.getMessage() + ":" + valueClass + ","
//						+ Arrays.toString(Iterables.toArray(keySet, String.class)), e);
//			} finally {
//				lock.unlock();
//			}
//		}
//		return result;
//	}

	@Override
	public Map<String, V> getAll(Iterable<? extends String> keys) {
		Set<String> keySet = Sets.newHashSet(keys);
		Map<String, V> result = redissonClient.getBuckets().get(Iterables.toArray(keys, keyClass));
		keySet.removeAll(result.keySet());
		if (!keySet.isEmpty()) {
			try {
				Set<String> queryKeys = Sets.newHashSet();// 需要从db获取
				Set<String> reGetKeys = Sets.newHashSet();// 需要再次从redis获取
				for (String key : keySet) {
					if (!redissonClient.getBucket(key).isExists()) {
						queryKeys.add(key);
					} else {
						reGetKeys.add(key);
					}
				}
				if (!queryKeys.isEmpty()) {
					List<V> queryList = persister.queryBatch(queryKeys);
					if (!queryList.isEmpty()) {
						RBatch batch = redissonClient.createBatch(BatchOptions.defaults());
						result.putAll(queryList.stream().collect(Collectors.toMap(V::getId, n -> {
							if (timeToLive > 0) {
								batch.getBucket(n.getId()).trySetAsync(n, timeToLive, timeToLiveUnit);
							} else {
								batch.getBucket(n.getId()).trySetAsync(n);
							}
							return n;
						}, (n, m) -> m)));
						
						batch.execute();
					}
				}
				if (!reGetKeys.isEmpty()) {
					Map<String, V> reGetMap = redissonClient.getBuckets()
							.get(Iterables.toArray(reGetKeys, getKeyClass()));
					result.putAll(reGetMap);
				}
			} catch (Exception e) {
				log.error(e.getMessage() + ":" + valueClass + ","
						+ Arrays.toString(Iterables.toArray(keySet, String.class)), e);
			}
		}
		return result;
	}
	
	@Override
	public boolean containsKey(Object key) {
		return redissonClient.getBucket((String) key).isExists() || persister.contains((String) key);
	}

	@Override
	public boolean containsValue(Object value) {
		return get(((V) value).getId()).equals(value);
	}

	@Override
	public V get(Object key) {
		return penetrateQuery((String) key, false);
	}

	@Override
	public V put(String key, V value) {
		try {
			if (timeToLive > 0)
				return (V) redissonClient.getBucket(key).getAndSet(value, timeToLive, timeToLiveUnit);
			else
				return (V) redissonClient.getBucket(key).getAndSet(value);
		} finally {
			synchronizer.flush(key, value);
		}
	}

	@Override
	public V remove(Object key) {
		try {
			return (V) redissonClient.getBucket((String) key).getAndDelete();
		} finally {
			synchronizer.giveUp((String) key);
			persister.delete((String) key);
			if (keyRecycler != null)
				keyRecycler.retrieve((String) key);
		}
	}

	@Override
	public void removeAll(Iterable<? extends String> keys) {
		redissonClient.getKeys().delete(Iterables.toArray(keys, keyClass));
		synchronizer.giveUp(keys);
		persister.delete(keys);
		if (keyRecycler != null)
			keyRecycler.retrieve(Iterables.toArray(keys, keyClass));
	}

	@Override
	public void putAll(Map<? extends String, ? extends V> m) {
		RBatch batch = redissonClient.createBatch(BatchOptions.defaults());
		m.forEach((key, value) -> {
			if (timeToLive > 0) {
				batch.getBucket(key).setAsync(value, timeToLive, timeToLiveUnit);
			} else {
				batch.getBucket(key).setAsync(value);
			}
		});
		batch.execute();
		synchronizer.flush(m);
	}

	@Override //
	public V putIfAbsent(String key, V value) {
		return RedissonAtomicExecutor.spinCall(redissonClient, key, lockWaitTime(), lockLeaseTime(), lockTimeUnit(),
				() -> {
					V v = penetrateQuery(key, false);
					if (v == null) {
						v = put(key, value);
					}
					return v;
				});
	}

	@Override //
	public boolean remove(Object key, Object value) {
		AtomicBoolean bool = new AtomicBoolean();
		RedissonAtomicExecutor.spinRun(redissonClient, (String) key, lockWaitTime(), lockLeaseTime(), lockTimeUnit(),
				() -> {
					Object curValue = penetrateQuery((String) key, false);
					if (!Objects.equals(curValue, value) || (curValue == null && !containsKey(key))) {

					} else {
						remove(key);
						bool.compareAndSet(false, true);
					}
				});
		return bool.get();
	}

	@Override //
	public boolean replace(String key, V oldValue, V newValue) {
		AtomicBoolean bool = new AtomicBoolean();
		RedissonAtomicExecutor.spinRun(redissonClient, (String) key, lockWaitTime(), lockLeaseTime(), lockTimeUnit(),
				() -> {
					Object curValue = penetrateQuery(key, false);
					if (!Objects.equals(curValue, oldValue) || (curValue == null && !containsKey(key))) {

					} else {
						put(key, newValue);
						bool.compareAndSet(false, true);
					}
				});
		return bool.get();
	}

	@Override //
	public V replace(String key, V value) {
		return RedissonAtomicExecutor.spinCall(redissonClient, key, lockWaitTime(), lockLeaseTime(), lockTimeUnit(),
				() -> {
					V curValue;
					if (((curValue = penetrateQuery(key, false)) != null) || containsKey(key)) {
						curValue = put(key, value);
					}
					return curValue;
				});
	}

	@Override //
	public V computeIfAbsent(String key, Function<? super String, ? extends V> mappingFunction) {
		return RedissonAtomicExecutor.spinCall(redissonClient, key, lockWaitTime(), lockLeaseTime(), lockTimeUnit(),
				() -> {
					Objects.requireNonNull(mappingFunction);
					V v;
					if ((v = penetrateQuery(key, false)) == null) {
						V newValue;
						if ((newValue = mappingFunction.apply(key)) != null) {
							put(key, newValue);
							return newValue;
						}
					}

					return v;
				});
	}

	@Override //
	public V computeIfPresent(String key, BiFunction<? super String, ? super V, ? extends V> remappingFunction) {
		return RedissonAtomicExecutor.spinCall(redissonClient, key, lockWaitTime(), lockLeaseTime(), lockTimeUnit(),
				() -> {
					Objects.requireNonNull(remappingFunction);
					V oldValue;
					if ((oldValue = penetrateQuery(key, false)) != null) {
						V newValue = remappingFunction.apply(key, oldValue);
						if (newValue != null) {
							put(key, newValue);
							return newValue;
						} else {
							remove(key);
							return null;
						}
					} else {
						return null;
					}
				});
	}

	@Override //
	public V compute(String key, BiFunction<? super String, ? super V, ? extends V> remappingFunction) {
		return RedissonAtomicExecutor.spinCall(redissonClient, key, lockWaitTime(), lockLeaseTime(), lockTimeUnit(),
				() -> {
					Objects.requireNonNull(remappingFunction);
					V oldValue = penetrateQuery(key, false);

					V newValue = remappingFunction.apply(key, oldValue);
					if (newValue == null) {
						// delete mapping
						if (oldValue != null || containsKey(key)) {
							// something to remove
							remove(key);
							return null;
						} else {
							// nothing to do. Leave things as they were.
							return null;
						}
					} else {
						// add or replace old mapping
						put(key, newValue);
						return newValue;
					}
				});
	}

	@Override //
	public V merge(String key, V value, BiFunction<? super V, ? super V, ? extends V> remappingFunction) {
		return RedissonAtomicExecutor.spinCall(redissonClient, key, lockWaitTime(), lockLeaseTime(), lockTimeUnit(),
				() -> {
					Objects.requireNonNull(remappingFunction);
					Objects.requireNonNull(value);
					V oldValue = penetrateQuery(key, false);
					V newValue = (oldValue == null) ? value : remappingFunction.apply(oldValue, value);
					if (newValue == null) {
						remove(key);
					} else {
						put(key, newValue);
					}
					return newValue;
				});
	}

	private V penetrateQuery(String key, boolean sync) {
		V value;
		if ((value = (V) redissonClient.getBucket(key).get()) == null) {
			value = persister.query(key);
			if (value != null) {
				boolean trySet;
				if (timeToLive > 0) {
					trySet = redissonClient.getBucket(key).trySet(value, timeToLive, timeToLiveUnit);
				} else {
					trySet = redissonClient.getBucket(key).trySet(value);
				}
				if (sync && trySet) {
					synchronizer.flush(key, value);
				}
			}
		}
		return value;
	}

	protected String keyPattern() {
		return RKeys.createRId(keyPrefix, "*");
	}

}
